package com.shujia.flink.tf

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo11Process {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * process: 在DataStream后面使用，
     * 是flink一个底层的算子，可以用于代替map,flatMap filter
     */
    val wordsDS: DataStream[(String, Int)] = linesDS.process(new ProcessFunction[String, (String, Int)] {
      /**
       * processElement:每一条数据执行一次， 进来一条可以返回多条
       *
       * @param value ： 一行数据
       * @param ctx   ：上下文对象
       * @param out   ： 用于将数据发送到下游,可以发送多条数据到下游
       */
      override def processElement(value: String,
                                  ctx: ProcessFunction[String, (String, Int)]#Context,
                                  out: Collector[(String, Int)]): Unit = {
        for (word <- value.split(",")) {
          //将数据发送到西游
          out.collect((word, 1))
        }
      }
    })

    wordsDS.print()

    env.execute()
  }

}
